package com.shujia.flink.window

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{SlidingEventTimeWindows, SlidingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time

import java.lang

object Demo1TimeWindow {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * 解析数据
     * 通过fastjson解析json格式的数据
     */
    val cardAndTimeDS: DataStream[(Long, Long)] = linesDS.map(car => {
      val jsonObj: JSONObject = JSON.parseObject(car)
      val card: Long = jsonObj.getLong("card")
      val time: Long = jsonObj.getLong("time") * 1000
      (card, time)
    })

    /**
     * 时间窗口
     * 1、SlidingProcessingTimeWindows：滑动的处理时间窗口
     * 2、SlidingEventTimeWindows：滑动的事件时间窗口
     * 3、TumblingProcessingTimeWindows：滚动的处理时间窗口
     * 4、TumblingEventTimeWindows：滚动的事件时间窗口
     *
     */

    /**
     * 处理时间窗口
     */
    /*    cardAndTimeDS
          .map(kv => (kv._1, 1))
          .keyBy(_._1)
          .window(SlidingProcessingTimeWindows.of(Time.seconds(15), Time.seconds(5)))
          .sum(1)
          .print()*/

    /**
     * 事件时间窗口 -- 需要设置时间字段和水位线
     */
    cardAndTimeDS
      //设置时间字段
      .assignAscendingTimestamps(_._2)
      .map(kv => (kv._1, 1))
      .keyBy(_._1)
      .window(SlidingEventTimeWindows.of(Time.seconds(15), Time.seconds(5)))
      .sum(1)
      .print()



    env.execute()

  }

}
